package msg

import (
	"dmsv/process"
	"encoding/json"
	"fmt"
	"github.com/beego/beego/v2/core/logs"
	"github.com/go-redis/redis"
	"strings"
	"time"
)

const (
	redisSyncChannel = "ipcc_sync"
)

type handleInfo struct {
	client      *redis.Client
	msgChannel  *redis.PubSub
	sendChannel chan *msgInfo
	localAddr   string
	dsid        int
}

func newHandler(passwd, addr, localIp string, dsid int) *handleInfo {

	c := redis.NewClient(&redis.Options{
		Addr:         addr,
		Password:     passwd,
		PoolSize:     10,
		MinIdleConns: 2,
		IdleTimeout:  300 * time.Second,
		DB:           6,
	})

	resChannel := c.Subscribe(redisSyncChannel)
	err := resChannel.Ping()
	if err == nil {

		res := &handleInfo{
			client:      c,
			localAddr:   localIp,
			dsid:        dsid,
			msgChannel:  resChannel,
			sendChannel: make(chan *msgInfo, 1000),
		}

		go res.receiver()

		return res
	}

	logs.Error("连接Redis 失败:%v", err)
	return nil
}

func (h *handleInfo) receiver() {

	for {

		ev, err := h.msgChannel.ReceiveMessage()
		if err != nil {
			logs.Error("receive error:%v", err)
			time.Sleep(2 * time.Second)
			continue
		}

		logs.Debug("receive new message:%v", ev.Payload)
		message := msgInfo{}
		err = json.Unmarshal([]byte(ev.Payload), &message)
		if err != nil {
			logs.Error("msg[%v] json decode failed:%v", ev.Payload, err)
			continue
		}

		if message.Sender == h.dsid {
			continue
		}

		syncToLocal(message.Event, message.Data)
	}
}

func syncToLocal(msgType int, data string) {

	switch msgType {

	case MSG_TYPE_COMPANY_ADD:

		info := MSG_SYNC_APP{}
		err := json.Unmarshal([]byte(data), &info)
		if err != nil {
			logs.Error("[%v] json err:%v", data, err)
			return
		}

		logs.Debug("同步添加企业:%v", info.CompanyId)
		appId := fmt.Sprintf("%d", info.CompanyId)
		_, err = process.AddNewEnterprice(appId, info.NavAudio, info.Number, info.Acd, info.GetKey)
		if err != nil {
			logs.Error("同步App[%v]失败:%v ", appId, err)
		}
	case MSG_TYPE_COMPANY_UPDATE:

		info := MSG_SYNC_APP{}
		err := json.Unmarshal([]byte(data), &info)
		if err != nil {
			logs.Error("[%v] json err:%v", data, err)
			return
		}

		logs.Debug("同步更新企业信息:%v", info.CompanyId)
		appId := fmt.Sprintf("%d", info.CompanyId)
		process.EnterpriceUpdate(appId, info.NavAudio, info.Acd, info.Number)

	case MSG_TYPE_AGENT_SIGN_IN:

		info := MSG_SYNC_SIGN{}
		err := json.Unmarshal([]byte(data), &info)
		if err != nil {
			logs.Error("[%v] json err:%v", data, err)
			return
		}

		companyId := fmt.Sprintf("%d", info.CompanyId)
		err = process.ServiceSignIn(companyId, info.QueId, info.Number, info.ServiceId, info.Display)
		if err != nil {
			logs.Error("同步签入信息失败:%v,%v", err.Error(), data)
		}

	case MSG_TYPE_AGENT_SIGN_OFF:

		info := MSG_SYNC_SIGN{}
		err := json.Unmarshal([]byte(data), &info)
		if err != nil {
			logs.Error("[%v] json err:%v", data, err)
			return
		}

		companyId := fmt.Sprintf("%d", info.CompanyId)
		err = process.ServiceSignOut(companyId, info.QueId, info.ServiceId)
		if err != nil {
			logs.Error("同步签出信息失败:%v,%v", err.Error(), data)
		}

	case MSG_TYPE_QUEUE_ADD:
		info := MSG_SYNC_QUE{}
		err := json.Unmarshal([]byte(data), &info)
		if err != nil {
			logs.Error("[%v] json err:%v", data, err)
			return
		}

		logs.Debug("同步添加队列:%v->%v,%v", info.CompanyId, info.Name, data)
		appId := fmt.Sprintf("%d", info.CompanyId)
		res := process.AddNewQue(appId, info.Name, info.Key, info.Media, info.Distrb, info.Id)
		if !res {
			logs.Error("add new Que failed:%v", data)
		}

	case MSG_TYPE_QUEUE_DEL:
		info := MSG_SYNC_QUE{}
		err := json.Unmarshal([]byte(data), &info)
		if err != nil {
			logs.Error("[%v] json err:%v", data, err)
			return
		}

		logs.Debug("同步删除队列:%v->%v,%v", info.CompanyId, info.Name, data)
		appId := fmt.Sprintf("%d", info.CompanyId)
		process.DelQue(appId, info.Name)

	case MSG_TYPE_QUEUE_UPDATE:
		info := MSG_SYNC_QUE{}
		err := json.Unmarshal([]byte(data), &info)
		if err != nil {
			logs.Error("[%v] json err:%v", data, err)
			return
		}

		logs.Debug("同步更新队列:%v->%v,%v", info.CompanyId, info.Name, data)
		appId := fmt.Sprintf("%d", info.CompanyId)
		res := process.UpdateQue(appId, info.Name, info.Key, info.Media, info.Distrb, info.Id)
		if false == res {
			logs.Debug("同步更新队列失败:%v->%v", info.CompanyId, info.Name)
		}

	case MSG_TYPE_SIP_ADD:
	case MSG_TYPE_SIP_UPDATE:
	case MSG_TYPE_SIP_DEL:

	default:
		logs.Error("wrong msg type :%v,drop it:%v", msgType, data)
	}
}

func (h *handleInfo) Send(data interface{}) error {

	res := h.client.Publish(redisSyncChannel, data)
	if res.Err() != nil {
		logs.Error("publish msg err:%v", res.Err())
		return res.Err()
	}

	logs.Debug("publish [%v] success,receiver:%v", string(data.([]byte)), res.Val())
	return nil
}

func (h *handleInfo) LockSet(lockName string) bool {

	res := h.client.SetNX(lockName, h.localAddr, time.Second*30)

	logs.Debug("get lock:%v ", res.String())
	if res.Val() || strings.Contains(res.String(), h.localAddr) {
		return true
	}

	return false
}
